package concurrent.push;

import java.util.*;
import java.util.concurrent.BlockingDeque;

/**
 * Author: Jason Wu
 * Date  : 2013-12-20
 */
public class Executor {
    private final BlockingDeque<PushingTask> taskQueue;
    private int threadCount;
    private int maxQueueSize;
    private List<Worker> threads;
    private QueueManager queueManager;

    public Executor(BlockingDeque<PushingTask> taskQueue, int threadCount, QueueManager queueManager) {
        this.taskQueue = taskQueue;
        this.threadCount = threadCount;
        this.maxQueueSize = Integer.MAX_VALUE;
        this.queueManager = queueManager;
        threads = new ArrayList<Worker>();
        for (int i = 0; i < threadCount; i++) {
            Worker worker = new Worker(i);
            threads.add(i, worker);
            worker.start();
        }
    }

    public Executor(BlockingDeque<PushingTask> taskQueue, int threadCount, int maxQueueSize, QueueManager queueManager) {
        this.taskQueue = taskQueue;
        this.threadCount = threadCount;
        this.maxQueueSize = maxQueueSize;
        this.queueManager = queueManager;
        threads = new ArrayList<Worker>();
        for (int i = 0; i < threadCount; i++) {
            Worker worker = new Worker(i);
            threads.add(i, worker);
            worker.start();
        }
    }

    public void execute(PushingTask task) {
        synchronized (taskQueue) {
            if (taskQueue.size() > maxQueueSize) {
                throw new RejectionException(String.format("Queue size out of MaxQueueSize[%d].", maxQueueSize));
            }
            taskQueue.addLast(task);
            taskQueue.notify();
            queueManager.add(task);
        }
    }

    private class Worker extends Thread {

        private boolean running = true;
        private int index;

        private Worker(int index) {
            this.index = index;
        }

        @Override
        public void run() {
            System.out.println("running:" + index);
            while (running) {
                PushingTask task = null;
                synchronized (taskQueue) {
                    try {
                        if (taskQueue.isEmpty()) {
                            taskQueue.wait();
                        }
                        task = taskQueue.take();
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                }
                try {
                    if (task == null || task.isCancelled()) {
                        continue;
                    }
                    task.push(index);
                } catch (Exception ex) {
                    ex.printStackTrace();
                } finally {
                    if (task != null) {
                        queueManager.remove(task);
                    }
                }
            }
        }

        public boolean shutdown() {
            running = false;
            return running;
        }

    }

    public synchronized void updateThread(int numberOfThread) {
        if (numberOfThread == threadCount) {
            return;
        }
        int offset = Math.abs(numberOfThread - threadCount);
        if (numberOfThread > threadCount) {
            for (int i = 0; i < offset; i++) {
                Worker worker = new Worker(threads.size());
                threads.add(worker);
                worker.start();
            }
        } else {
            for (int i = 0; i < offset; i++) {
                Worker worker = threads.remove(threads.size() - 1);
                worker.shutdown();
            }
        }
        threadCount = numberOfThread;
    }

    public synchronized void updateQueueSize(int queueSize) {
        synchronized (taskQueue) {
            maxQueueSize = queueSize;
        }
    }

}
